/**
 * Copyright © 2013 Creativor Studio.
 * About license information, please see LICENSE.txt.
 */
package org.creativor.rayson.client.transport.nio;

import java.io.IOException;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.creativor.rayson.share.transport.nio.NioConnection;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Client side socket listener. NIO implement.
 * 
 * @author creativor
 */
public class NioListener extends Thread {
	private static final Logger LOG = LoggerFactory
			.getLogger(NioListener.class);
	private final static NioListener SINGLETON = new NioListener();

	/**
	 * @return Singleton listener.
	 */
	public static NioListener getListener() {
		return SINGLETON;
	}

	private final Lock lock;

	private final Condition nonePendingTasks;
	private long pendingTaskCount;

	private boolean running = true;

	private Selector selector;

	private NioListener() {
		setName("NIO client listener of Rayson");
		setDaemon(true);
		this.lock = new ReentrantLock();
		this.nonePendingTasks = lock.newCondition();
		this.pendingTaskCount = 0;

	}

	/**
	 * Accept a new connection.
	 * 
	 * @param socketChannel
	 *            Underling socket channel.
	 * @param conn
	 *            The client connection to be accept.
	 * @return selection key register into the selector of this listener.
	 * @throws ClosedChannelException
	 *             If the underling socket channel is closed before the result
	 *             is return .
	 */
	SelectionKey accept(final SocketChannel socketChannel,
			final NioConnection conn) throws ClosedChannelException {
		SelectionKey key;
		selector.wakeup();
		lock.lock();
		try {
			pendingTaskCount++;
			try {
				key = socketChannel.register(selector, SelectionKey.OP_READ);
				key.attach(conn);
			} finally {
				pendingTaskCount--;
			}
			if (pendingTaskCount == 0)
				nonePendingTasks.signalAll();
		} finally {
			lock.unlock();
		}
		LOG.info(conn + " accepted");
		return key;
	}

	/**
	 * Handling read event of this listener.
	 * 
	 * @param key
	 */
	private void doRead(final SelectionKey key) {
		final NioConnection connection = (NioConnection) key.attachment();
		int readCount = -1;
		try {
			readCount = connection.asyncRead();
		} catch (final IOException e) {
		}
		if (readCount == -1) {
			try {
				connection.close();
			} catch (final IOException e) {
				// Do nothing about this exception.
			}
			LOG.info("connection " + connection + " closed, by EOF exception");
		}

	}

	/**
	 * Handling write event of this listener.
	 * 
	 * @param key
	 */
	private void doWrite(final SelectionKey key) {
		final NioConnection connection = (NioConnection) key.attachment();
		try {
			connection.asyncWrite();
		} catch (final IOException e) {
			try {
				connection.close();
			} catch (final IOException e1) {
				// Do nothing about this exception.
			}
			LOG.info("connection " + connection + " closed, by write exception");
		}
	}

	/**
	 * Initialize this listener, also start this thread.
	 * 
	 * @throws IOException
	 *             If can not open the selector.
	 */
	public void initialize() throws IOException {
		this.selector = Selector.open();
		this.start();
	}

	/**
	 * Quit this listener thread.
	 */
	void quit() {
		this.running = false;
	}

	@Override
	public void run() {
		LOG.info(getName() + " starting ..." + "");
		SelectionKey key;
		Iterator<SelectionKey> iterator;

		while (running) {
			try {
				selector.select();
				lock.lock();
				try {
					// try to wait pending accept tasks.
					while (pendingTaskCount > 0) {
						nonePendingTasks.await();
					}
					for (iterator = selector.selectedKeys().iterator(); iterator
							.hasNext();) {

						key = iterator.next();
						iterator.remove();
						if (key.isValid()) {
							if (key.isReadable())
								doRead(key);
							else if (key.isWritable())
								doWrite(key);
							else {
								LOG.error("Listener got un-excepted key: "
										+ key.toString());
							}
						}
					}
				} finally {
					lock.unlock();
				}
			} catch (final Throwable e) {
				// protect this thread not to quit.
				LOG.error("Server listener got error", e);
			}

		}
		LOG.info(getName() + " stopped");
	}
}